// Copyright (C) Kumo inc. and its affiliates.
// Author: Jeff.li lijippy@163.com
// All rights reserved.
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published
// by the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <https://www.gnu.org/licenses/>.
//

#pragma once

#include <atomic>
#include <cmath>
#include <functional>
#include <memory>
#include <optional>
#include <type_traits>
#include <utility>
#include <vector>

#include <turbo/log/logging.h>
#include <turbo/utility/status.h>
#include <nebula/version.h>
#include <turbo/functional/functional.h>
#include <turbo/base/macros.h>
#include <nebula/trace/tracing.h>
#include <turbo/base/macros.h>
#include <nebula/future/type_fwd.h>

namespace nebula {

    template<typename>
    struct EnsureFuture;

    namespace detail {

        template<typename>
        struct is_future : std::false_type {
        };

        template<typename T>
        struct is_future<Future<T>> : std::true_type {
        };

        template<typename Signature, typename Enable = void>
        struct result_of;

        template<typename Fn, typename... A>
        struct result_of<Fn(A...),
                std::void_t<decltype(std::declval<Fn>()(std::declval<A>()...))>> {
            using type = decltype(std::declval<Fn>()(std::declval<A>()...));
        };

        template<typename Signature>
        using result_of_t = typename result_of<Signature>::type;

        // Helper to find the synchronous counterpart for a Future
        template<typename T>
        struct SyncType {
            using type = turbo::Result<T>;
        };

        template<>
        struct SyncType<turbo::EmptyResult> {
            using type = turbo::Status;
        };

        template<typename Fn>
        using first_arg_is_status =
                std::is_same<typename std::decay<turbo::call_traits::argument_type<0, Fn>>::type,
                        turbo::Status>;

        template<typename Fn, typename Then, typename Else,
                typename Count = turbo::call_traits::argument_count<Fn>>
        using if_has_no_args = typename std::conditional<Count::value == 0, Then, Else>::type;

        /// Creates a callback that can be added to a future to mark a `dest` future finished
        template<typename Source, typename Dest, bool SourceEmpty = Source::is_empty,
                bool DestEmpty = Dest::is_empty>
        struct MarkNextFinished {
        };

        /// If the source and dest are both empty we can pass on the status
        template<typename Source, typename Dest>
        struct MarkNextFinished<Source, Dest, true, true> {
            void operator()(const turbo::Status &status) &&{ next.mark_finished(status); }

            Dest next;
        };

        /// If the source is not empty but the dest is then we can take the
        /// status out of the result
        template<typename Source, typename Dest>
        struct MarkNextFinished<Source, Dest, false, true> {
            void operator()(const turbo::Result<typename Source::value_type> &res) &&{
                next.mark_finished(turbo::EmptyResult::to_result(res.status()));
            }

            Dest next;
        };

        /// If neither are empty we pass on the result
        template<typename Source, typename Dest>
        struct MarkNextFinished<Source, Dest, false, false> {
            void operator()(const turbo::Result<typename Source::value_type> &res) &&{
                next.mark_finished(res);
            }

            Dest next;
        };

        /// Helper that contains information about how to apply a continuation
        struct ContinueFuture {
            template<typename Return>
            struct ForReturnImpl;

            template<typename Return>
            using ForReturn = typename ForReturnImpl<Return>::type;

            template<typename Signature>
            using ForSignature = ForReturn<result_of_t<Signature>>;

            // If the callback returns void then we return Future<> that always finishes OK.
            template<typename ContinueFunc, typename... Args,
                    typename ContinueResult = result_of_t<ContinueFunc &&(Args &&...)>,
                    typename NextFuture = ForReturn<ContinueResult>>
            typename std::enable_if<std::is_void<ContinueResult>::value>::type operator()(
                    NextFuture next, ContinueFunc &&f, Args &&... a) const {
                std::forward<ContinueFunc>(f)(std::forward<Args>(a)...);
                next.mark_finished();
            }

            /// If the callback returns a non-future then we return Future<T>
            /// and mark the future finished with the callback result.  It will get promoted
            /// to turbo::Result<T> as part of mark_finished if it isn't already.
            ///
            /// If the callback returns turbo::Status and we return Future<> then also send the callback
            /// result as-is to the destination future.
            template<typename ContinueFunc, typename... Args,
                    typename ContinueResult = result_of_t<ContinueFunc &&(Args &&...)>,
                    typename NextFuture = ForReturn<ContinueResult>>
            typename std::enable_if<
                    !std::is_void<ContinueResult>::value && !is_future<ContinueResult>::value &&
                    (!NextFuture::is_empty || std::is_same<ContinueResult, turbo::Status>::value)>::type
            operator()(NextFuture next, ContinueFunc &&f, Args &&... a) const {
                next.mark_finished(std::forward<ContinueFunc>(f)(std::forward<Args>(a)...));
            }

            /// If the callback returns a turbo::Result and the next future is Future<> then we mark
            /// the future finished with the callback result.
            ///
            /// It may seem odd that the next future is Future<> when the callback returns a
            /// result but this can occur if the OnFailure callback returns a result while the
            /// OnSuccess callback is void/turbo::Status (e.g. you would get this calling the one-arg
            /// version of Then with an OnSuccess callback that returns void)
            template<typename ContinueFunc, typename... Args,
                    typename ContinueResult = result_of_t<ContinueFunc &&(Args &&...)>,
                    typename NextFuture = ForReturn<ContinueResult>>
            typename std::enable_if<!std::is_void<ContinueResult>::value &&
                                    !is_future<ContinueResult>::value && NextFuture::is_empty &&
                                    !std::is_same<ContinueResult, turbo::Status>::value>::type
            operator()(NextFuture next, ContinueFunc &&f, Args &&... a) const {
                next.mark_finished(std::forward<ContinueFunc>(f)(std::forward<Args>(a)...).status());
            }

            /// If the callback returns a Future<T> then we return Future<T>.  We create a new
            /// future and add a callback to the future given to us by the user that forwards the
            /// result to the future we just created
            template<typename ContinueFunc, typename... Args,
                    typename ContinueResult = result_of_t<ContinueFunc &&(Args &&...)>,
                    typename NextFuture = ForReturn<ContinueResult>>
            typename std::enable_if<is_future<ContinueResult>::value>::type operator()(
                    NextFuture next, ContinueFunc &&f, Args &&... a) const {
                ContinueResult signal_to_complete_next =
                        std::forward<ContinueFunc>(f)(std::forward<Args>(a)...);
                MarkNextFinished<ContinueResult, NextFuture> callback{std::move(next)};
                signal_to_complete_next.add_callback(std::move(callback));
            }

            /// Helpers to conditionally ignore arguments to ContinueFunc
            template<typename ContinueFunc, typename NextFuture, typename... Args>
            void IgnoringArgsIf(std::true_type, NextFuture &&next, ContinueFunc &&f,
                                Args &&...) const {
                operator()(std::forward<NextFuture>(next), std::forward<ContinueFunc>(f));
            }

            template<typename ContinueFunc, typename NextFuture, typename... Args>
            void IgnoringArgsIf(std::false_type, NextFuture &&next, ContinueFunc &&f,
                                Args &&... a) const {
                operator()(std::forward<NextFuture>(next), std::forward<ContinueFunc>(f),
                           std::forward<Args>(a)...);
            }
        };

        /// Helper struct which tells us what kind of Future gets returned from `Then` based on
        /// the return type of the OnSuccess callback
        template<>
        struct ContinueFuture::ForReturnImpl<void> {
            using type = Future<>;
        };

        template<>
        struct ContinueFuture::ForReturnImpl<turbo::Status> {
            using type = Future<>;
        };

        template<typename R>
        struct ContinueFuture::ForReturnImpl {
            using type = Future<R>;
        };

        template<typename T>
        struct ContinueFuture::ForReturnImpl<turbo::Result<T>> {
            using type = Future<T>;
        };

        template<typename T>
        struct ContinueFuture::ForReturnImpl<Future<T>> {
            using type = Future<T>;
        };

    }  // namespace detail

    /// A Future's execution or completion status
    enum class FutureState : int8_t {
        PENDING, SUCCESS, FAILURE
    };

    inline bool IsFutureFinished(FutureState state) { return state != FutureState::PENDING; }

    /// \brief Describe whether the callback should be scheduled or run synchronously
    enum class ShouldSchedule {
        /// Always run the callback synchronously (the default)
        Never = 0,
        /// schedule a new task only if the future is not finished when the
        /// callback is added
        IfUnfinished = 1,
        /// Always schedule the callback as a new task
        Always = 2,
        /// schedule a new task only if it would run on an executor other than
        /// the specified executor.
        IfDifferentExecutor = 3,
    };

    /// \brief Options that control how a continuation is run
    struct CallbackOptions {
        /// Describe whether the callback should be run synchronously or scheduled
        ShouldSchedule should_schedule = ShouldSchedule::Never;
        /// If the callback is scheduled then this is the executor it should be scheduled
        /// on.  If this is NULL then should_schedule must be Never
        internal::Executor *executor = nullptr;

        static CallbackOptions defaults() { return {}; }
    };

    // Untyped private implementation
    class TURBO_EXPORT FutureImpl : public std::enable_shared_from_this<FutureImpl> {
    public:
        FutureImpl();

        virtual ~FutureImpl() = default;

        FutureState state() { return state_.load(); }

        static std::unique_ptr<FutureImpl> create();

        static std::unique_ptr<FutureImpl> make_finished(FutureState state);

#ifdef NEBULA_WITH_OPENTELEMETRY
        void SetSpan(util::tracing::Span* span) { span_ = span; }
#endif

        // Future API
        void mark_finished();

        void MarkFailed();

        void Wait();

        bool Wait(double seconds);

        template<typename VT>
        turbo::Result<VT> *CastResult() const {
            return static_cast<turbo::Result<VT> *>(result_.get());
        }

        using Callback = turbo::FnOnce<void(const FutureImpl &impl)>;

        void add_callback(Callback callback, CallbackOptions opts);

        bool try_add_callback(const std::function<Callback()> &callback_factory,
                            CallbackOptions opts);

        std::atomic<FutureState> state_{FutureState::PENDING};

        // Type erased storage for arbitrary results
        // XXX small objects could be stored inline instead of boxed in a pointer
        using Storage = std::unique_ptr<void, void (*)(void *)>;
        Storage result_{nullptr, nullptr};

        struct CallbackRecord {
            Callback callback;
            CallbackOptions options;
        };
        std::vector<CallbackRecord> callbacks_;
#ifdef NEBULA_WITH_OPENTELEMETRY
        util::tracing::Span* span_ = nullptr;
#endif
    };

    // ---------------------------------------------------------------------
    // Public API

    /// \brief EXPERIMENTAL A std::future-like class with more functionality.
    ///
    /// A Future represents the results of a past or future computation.
    /// The Future API has two sides: a producer side and a consumer side.
    ///
    /// The producer API allows creating a Future and setting its result or
    /// status, possibly after running a computation function.
    ///
    /// The consumer API allows querying a Future's current state, wait for it
    /// to complete, and composing futures with callbacks.
    template<typename T>
    class [[nodiscard]] Future {
    public:
        using value_type = T;
        using SyncType = typename detail::SyncType<T>::type;
        static constexpr bool is_empty = std::is_same<T, turbo::EmptyResult>::value;

        // The default constructor creates an invalid Future.  Use Future::create()
        // for a valid Future.  This constructor is mostly for the convenience
        // of being able to presize a vector of Futures.
        Future() = default;

#ifdef NEBULA_WITH_OPENTELEMETRY
        void SetSpan(util::tracing::Span* span) { impl_->SetSpan(span); }
#endif

        // Consumer API

        bool is_valid() const { return impl_ != nullptr; }

        /// \brief Return the Future's current state
        ///
        /// A return value of PENDING is only indicative, as the Future can complete
        /// concurrently.  A return value of FAILURE or SUCCESS is definitive, though.
        FutureState state() const {
            check_valid();
            return impl_->state();
        }

        /// \brief Whether the Future is finished
        ///
        /// A false return value is only indicative, as the Future can complete
        /// concurrently.  A true return value is definitive, though.
        bool is_finished() const {
            check_valid();
            return IsFutureFinished(impl_->state());
        }

        /// \brief Wait for the Future to complete and return its turbo::Result
        const turbo::Result<value_type> &result() const &{
            Wait();
            return *get_result();
        }

        /// \brief Returns an rvalue to the result.  This method is potentially unsafe
        ///
        /// The future is not the unique owner of the result, copies of a future will
        /// also point to the same result.  You must make sure that no other copies
        /// of the future exist.  Attempts to add callbacks after you move the result
        /// will result in undefined behavior.
        turbo::Result<value_type> &&nove_result() {
            Wait();
            return std::move(*get_result());
        }

        /// \brief Wait for the Future to complete and return its turbo::Status
        const turbo::Status &status() const { return result().status(); }

        /// \brief Future<T> is convertible to Future<>, which views only the
        /// turbo::Status of the original. Marking the returned Future Finished is not supported.
        explicit operator Future<>() const {
            Future<> status_future;
            status_future.impl_ = impl_;
            return status_future;
        }

        /// \brief Wait for the Future to complete
        void Wait() const {
            check_valid();
            impl_->Wait();
        }

        /// \brief Wait for the Future to complete, or for the timeout to expire
        ///
        /// `true` is returned if the Future completed, `false` if the timeout expired.
        /// Note a `false` value is only indicative, as the Future can complete
        /// concurrently.
        bool Wait(double seconds) const {
            check_valid();
            return impl_->Wait(seconds);
        }

        // Producer API

        /// \brief Producer API: mark Future finished
        ///
        /// The Future's result is set to `res`.
        void mark_finished(turbo::Result<value_type> res) { do_mark_finished(std::move(res)); }

        /// \brief Mark a Future<> completed with the provided turbo::Status.
        template<typename E = value_type, typename = typename std::enable_if<
                std::is_same<E, turbo::EmptyResult>::value>::type>
        void mark_finished(turbo::Status s = turbo::OkStatus()) {
            return do_mark_finished(E::to_result(std::move(s)));
        }

        /// \brief Producer API: instantiate a valid Future
        ///
        /// The Future's state is initialized with PENDING.  If you are creating a future with
        /// this method you must ensure that future is eventually completed (with success or
        /// failure).  Creating a future, returning it, and never completing the future can lead
        /// to memory leaks (for example, see Loop).
        static Future create() {
            Future fut;
            fut.impl_ = FutureImpl::create();
            return fut;
        }

        /// \brief Producer API: instantiate a finished Future
        static Future<value_type> make_finished(turbo::Result<value_type> res) {
            Future<value_type> fut;
            fut.initialize_from_result(std::move(res));
            return fut;
        }

        /// \brief Make a finished Future<> with the provided turbo::Status.
        template<typename E = value_type, typename = typename std::enable_if<
                std::is_same<E, turbo::EmptyResult>::value>::type>
        static Future<> make_finished(turbo::Status s = turbo::OkStatus()) {
            return make_finished(E::to_result(std::move(s)));
        }

        struct WrapResultOnComplete {
            template<typename OnComplete>
            struct Callback {
                void operator()(const FutureImpl &impl) &&{
                    std::move(on_complete)(*impl.CastResult<value_type>());
                }

                OnComplete on_complete;
            };
        };

        struct WrapStatusyOnComplete {
            template<typename OnComplete>
            struct Callback {
                static_assert(std::is_same<turbo::EmptyResult, value_type>::value,
                              "Only callbacks for Future<> should accept turbo::Status and not turbo::Result");

                void operator()(const FutureImpl &impl) &&{
                    std::move(on_complete)(impl.CastResult<value_type>()->status());
                }

                OnComplete on_complete;
            };
        };

        template<typename OnComplete>
        using WrapOnComplete = typename std::conditional<
                detail::first_arg_is_status<OnComplete>::value, WrapStatusyOnComplete,
                WrapResultOnComplete>::type::template Callback<OnComplete>;

        /// \brief Consumer API: Register a callback to run when this future completes
        ///
        /// The callback should receive the result of the future (const turbo::Result<T>&)
        /// For a void or statusy future this should be (const turbo::Status&)
        ///
        /// There is no guarantee to the order in which callbacks will run.  In
        /// particular, callbacks added while the future is being marked complete
        /// may be executed immediately, ahead of, or even the same time as, other
        /// callbacks that have been previously added.
        ///
        /// WARNING: callbacks may hold arbitrary references, including cyclic references.
        /// Since callbacks will only be destroyed after they are invoked, this can lead to
        /// memory leaks if a Future is never marked finished (abandoned):
        ///
        /// {
        ///     auto fut = Future<>::create();
        ///     fut.add_callback([fut]() {});
        /// }
        ///
        /// In this example `fut` falls out of scope but is not destroyed because it holds a
        /// cyclic reference to itself through the callback.
        template<typename OnComplete, typename Callback = WrapOnComplete<OnComplete>>
        void add_callback(OnComplete on_complete,
                         CallbackOptions opts = CallbackOptions::defaults()) const {
            // We know impl_ will not be dangling when invoking callbacks because at least one
            // thread will be waiting for mark_finished to return. Thus it's safe to keep a
            // weak reference to impl_ here
            impl_->add_callback(Callback{std::move(on_complete)}, opts);
        }

        /// \brief Overload of add_callback that will return false instead of running
        /// synchronously
        ///
        /// This overload will guarantee the callback is never run synchronously.  If the future
        /// is already finished then it will simply return false.  This can be useful to avoid
        /// stack overflow in a situation where you have recursive Futures.  For an example
        /// see the Loop function
        ///
        /// Takes in a callback factory function to allow moving callbacks (the factory function
        /// will only be called if the callback can successfully be added)
        ///
        /// Returns true if a callback was actually added and false if the callback failed
        /// to add because the future was marked complete.
        template<typename CallbackFactory,
                typename OnComplete = detail::result_of_t<CallbackFactory()>,
                typename Callback = WrapOnComplete<OnComplete>>
        bool try_add_callback(CallbackFactory callback_factory,
                            CallbackOptions opts = CallbackOptions::defaults()) const {
            return impl_->try_add_callback([&]() { return Callback{callback_factory()}; }, opts);
        }

        template<typename OnSuccess, typename OnFailure>
        struct ThenOnComplete {
            static constexpr bool has_no_args =
                    turbo::call_traits::argument_count<OnSuccess>::value == 0;

            using ContinuedFuture = detail::ContinueFuture::ForSignature<
                    detail::if_has_no_args<OnSuccess, OnSuccess &&(), OnSuccess &&(const T &)>>;

            static_assert(
                    std::is_same<detail::ContinueFuture::ForSignature<OnFailure &&(const turbo::Status &)>,
                            ContinuedFuture>::value,
                    "OnSuccess and OnFailure must continue with the same future type");

            struct DummyOnSuccess {
                void operator()(const T &);
            };

            using OnSuccessArg = typename std::decay<turbo::call_traits::argument_type<
                    0, detail::if_has_no_args<OnSuccess, DummyOnSuccess, OnSuccess>>>::type;

            static_assert(
                    !std::is_same<OnSuccessArg, typename turbo::EnsureResult<OnSuccessArg>::type>::value,
                    "OnSuccess' argument should not be a turbo::Result");

            void operator()(const turbo::Result<T> &result) &&{
                detail::ContinueFuture continue_future;
                if (TURBO_LIKELY(result.ok())) {
                    // move on_failure to a(n immediately destroyed) temporary to free its resources
                    TURBO_UNUSED(OnFailure(std::move(on_failure)));
                    continue_future.IgnoringArgsIf(
                            detail::if_has_no_args<OnSuccess, std::true_type, std::false_type>{},
                            std::move(next), std::move(on_success), result.value_or_die());
                } else {
                    TURBO_UNUSED(OnSuccess(std::move(on_success)));
                    continue_future(std::move(next), std::move(on_failure), result.status());
                }
            }

            OnSuccess on_success;
            OnFailure on_failure;
            ContinuedFuture next;
        };

        template<typename OnSuccess>
        struct PassthruOnFailure {
            using ContinuedFuture = detail::ContinueFuture::ForSignature<
                    detail::if_has_no_args<OnSuccess, OnSuccess &&(), OnSuccess &&(const T &)>>;

            turbo::Result<typename ContinuedFuture::value_type> operator()(const turbo::Status &s) { return s; }
        };

        /// \brief Consumer API: Register a continuation to run when this future completes
        ///
        /// The continuation will run in the same thread that called mark_finished (whatever
        /// callback is registered with this function will run before mark_finished returns).
        /// Avoid long-running callbacks in favor of submitting a task to an Executor and
        /// returning the future.
        ///
        /// Two callbacks are supported:
        /// - OnSuccess, called with the result (const value_type&) on successful completion.
        ///              for an empty future this will be called with nothing ()
        /// - OnFailure, called with the error (const turbo::Status&) on failed completion.
        ///              This callback is optional and defaults to a passthru of any errors.
        ///
        /// Then() returns a Future whose value_type is derived from the return type of the
        /// callbacks. If a callback returns:
        /// - void, a Future<> will be returned which will completes successfully as soon
        ///   as the callback runs.
        /// - turbo::Status, a Future<> will be returned which will complete with the returned turbo::Status
        ///   as soon as the callback runs.
        /// - V or turbo::Result<V>, a Future<V> will be returned which will complete with the result
        ///   of invoking the callback as soon as the callback runs.
        /// - Future<V>, a Future<V> will be returned which will be marked complete when the
        ///   future returned by the callback completes (and will complete with the same
        ///   result).
        ///
        /// The continued Future type must be the same for both callbacks.
        ///
        /// Note that OnFailure can swallow errors, allowing continued Futures to successfully
        /// complete even if this Future fails.
        ///
        /// If this future is already completed then the callback will be run immediately
        /// and the returned future may already be marked complete.
        ///
        /// See add_callback for general considerations when writing callbacks.
        template<typename OnSuccess, typename OnFailure = PassthruOnFailure<OnSuccess>,
                typename OnComplete = ThenOnComplete<OnSuccess, OnFailure>,
                typename ContinuedFuture = typename OnComplete::ContinuedFuture>
        ContinuedFuture Then(OnSuccess on_success, OnFailure on_failure = {},
                             CallbackOptions options = CallbackOptions::defaults()) const {
            auto next = ContinuedFuture::create();
            add_callback(OnComplete{std::forward<OnSuccess>(on_success),
                                   std::forward<OnFailure>(on_failure), next},
                        options);
            return next;
        }

        /// \brief implicit constructor to create a finished future from a value
        Future(value_type val) : Future() {  // NOLINT runtime/explicit
            impl_ = FutureImpl::make_finished(FutureState::SUCCESS);
            set_result(std::move(val));
        }

        /// \brief implicit constructor to create a future from a turbo::Result, enabling use
        ///     of macros like TURBO_MOVE_OR_RAISE.
        Future(turbo::Result<value_type> res) : Future() {  // NOLINT runtime/explicit
            if (TURBO_LIKELY(res.ok())) {
                impl_ = FutureImpl::make_finished(FutureState::SUCCESS);
            } else {
                impl_ = FutureImpl::make_finished(FutureState::FAILURE);
            }
            set_result(std::move(res));
        }

        /// \brief implicit constructor to create a future from a turbo::Status, enabling use
        ///     of macros like TURBO_RETURN_NOT_OK.
        Future(turbo::Status s)  // NOLINT runtime/explicit
                : Future(turbo::Result<value_type>(std::move(s))) {}

    protected:
        void initialize_from_result(turbo::Result<value_type> res) {
            if (TURBO_LIKELY(res.ok())) {
                impl_ = FutureImpl::make_finished(FutureState::SUCCESS);
            } else {
                impl_ = FutureImpl::make_finished(FutureState::FAILURE);
            }
            set_result(std::move(res));
        }

        void Initialize() { impl_ = FutureImpl::create(); }

        turbo::Result<value_type> *get_result() const { return impl_->CastResult<value_type>(); }

        void set_result(turbo::Result<value_type> res) {
            impl_->result_ = {new turbo::Result<value_type>(std::move(res)),
                              [](void *p) { delete static_cast<turbo::Result<value_type> *>(p); }};
        }

        void do_mark_finished(turbo::Result<value_type> res) {
            set_result(std::move(res));

            if (TURBO_LIKELY(get_result()->ok())) {
                impl_->mark_finished();
            } else {
                impl_->MarkFailed();
            }
        }

        void check_valid() const {
#ifndef NDEBUG
            if (!is_valid()) {
                turbo::invalid_argument_error("Invalid Future (default-initialized?)").abort();
            }
#endif
        }

        explicit Future(std::shared_ptr<FutureImpl> impl) : impl_(std::move(impl)) {}

        std::shared_ptr<FutureImpl> impl_;

        friend struct detail::ContinueFuture;

        template<typename U>
        friend
        class Future;

        friend class WeakFuture<T>;

        FRIEND_TEST(FutureRefTest, ChainRemoved);

        FRIEND_TEST(FutureRefTest, TailRemoved);

        FRIEND_TEST(FutureRefTest, HeadRemoved);
    };

    template<typename T>
    typename Future<T>::SyncType FutureToSync(const Future<T> &fut) {
        return fut.result();
    }

    template<>
    inline typename Future<turbo::EmptyResult>::SyncType FutureToSync<turbo::EmptyResult>(
            const Future<turbo::EmptyResult> &fut) {
        return fut.status();
    }

    template<>
    inline Future<>::Future(turbo::Status s) : Future(turbo::EmptyResult::to_result(std::move(s))) {}

    template<typename T>
    class WeakFuture {
    public:
        explicit WeakFuture(const Future<T> &future) : impl_(future.impl_) {}

        Future<T> get() { return Future<T>{impl_.lock()}; }

    private:
        std::weak_ptr<FutureImpl> impl_;
    };

    /// \defgroup future-utilities Functions for working with Futures
    /// @{

    /// If a turbo::Result<Future> holds an error instead of a Future, construct a finished Future
    /// holding that error.
    template<typename T>
    static Future<T> DeferNotOk(turbo::Result<Future<T>> maybe_future) {
        if (TURBO_UNLIKELY(!maybe_future.ok())) {
            return Future<T>::make_finished(std::move(maybe_future).status());
        }
        return std::move(maybe_future).move_value_unsafe();
    }

    /// \brief create a Future which completes when all of `futures` complete.
    ///
    /// The future's result is a vector of the results of `futures`.
    /// Note that this future will never be marked "failed"; failed results
    /// will be stored in the result vector alongside successful results.
    template<typename T>
    Future<std::vector<turbo::Result<T>>> All(std::vector<Future<T>> futures) {
        struct State {
            explicit State(std::vector<Future<T>> f)
                    : futures(std::move(f)), n_remaining(futures.size()) {}

            std::vector<Future<T>> futures;
            std::atomic<size_t> n_remaining;
        };

        if (futures.size() == 0) {
            return {std::vector<turbo::Result<T>>{}};
        }

        auto state = std::make_shared<State>(std::move(futures));

        auto out = Future<std::vector<turbo::Result<T>>>::create();
        for (const Future<T> &future: state->futures) {
            future.add_callback([state, out](const turbo::Result<T> &) mutable {
                if (state->n_remaining.fetch_sub(1) != 1) return;

                std::vector<turbo::Result<T>> results(state->futures.size());
                for (size_t i = 0; i < results.size(); ++i) {
                    results[i] = state->futures[i].result();
                }
                out.mark_finished(std::move(results));
            });
        }
        return out;
    }

    /// \brief create a Future which completes when all of `futures` complete.
    ///
    /// The future will be marked complete if all `futures` complete
    /// successfully. Otherwise, it will be marked failed with the status of
    /// the first failing future.
    TURBO_EXPORT
    Future<> AllComplete(const std::vector<Future<>> &futures);

    /// \brief create a Future which completes when all of `futures` complete.
    ///
    /// The future will finish with an ok status if all `futures` finish with
    /// an ok status. Otherwise, it will be marked failed with the status of
    /// one of the failing futures.
    ///
    /// Unlike AllComplete this Future will not complete immediately when a
    /// failure occurs.  It will wait until all futures have finished.
    TURBO_EXPORT
    Future<> AllFinished(const std::vector<Future<>> &futures);

    /// @}

    struct Continue {
        template<typename T>
        operator std::optional<T>() &&{  // NOLINT explicit
            return {};
        }
    };

    template<typename T = turbo::EmptyResult>
    std::optional<T> Break(T break_value = {}) {
        return std::optional<T>{std::move(break_value)};
    }

    template<typename T = turbo::EmptyResult>
    using ControlFlow = std::optional<T>;

    /// \brief Loop through an asynchronous sequence
    ///
    /// \param[in] iterate A generator of Future<ControlFlow<BreakValue>>. On completion
    /// of each yielded future the resulting ControlFlow will be examined. A Break will
    /// terminate the loop, while a Continue will re-invoke `iterate`.
    ///
    /// \return A future which will complete when a Future returned by iterate completes with
    /// a Break
    template<typename Iterate,
            typename Control = typename detail::result_of_t<Iterate()>::value_type,
            typename BreakValueType = typename Control::value_type>
    Future<BreakValueType> Loop(Iterate iterate) {
        struct Callback {
            bool CheckForTermination(const turbo::Result<Control> &control_res) {
                if (!control_res.ok()) {
                    break_fut.mark_finished(control_res.status());
                    return true;
                }
                if (control_res->has_value()) {
                    break_fut.mark_finished(**control_res);
                    return true;
                }
                return false;
            }

            void operator()(const turbo::Result<Control> &maybe_control) &&{
                if (CheckForTermination(maybe_control)) return;

                auto control_fut = iterate();
                while (true) {
                    if (control_fut.try_add_callback([this]() { return *this; })) {
                        // Adding a callback succeeded; control_fut was not finished
                        // and we must wait to CheckForTermination.
                        return;
                    }
                    // Adding a callback failed; control_fut was finished and we
                    // can CheckForTermination immediately. This also avoids recursion and potential
                    // stack overflow.
                    if (CheckForTermination(control_fut.result())) return;

                    control_fut = iterate();
                }
            }

            Iterate iterate;

            // If the future returned by control_fut is never completed then we will be hanging on
            // to break_fut forever even if the listener has given up listening on it.  Instead we
            // rely on the fact that a producer (the caller of Future<>::create) is always
            // responsible for completing the futures they create.
            // TODO: Could avoid this kind of situation with "future abandonment" similar to mesos
            Future<BreakValueType> break_fut;
        };

        auto break_fut = Future<BreakValueType>::create();
        auto control_fut = iterate();
        control_fut.add_callback(Callback{std::move(iterate), break_fut});

        return break_fut;
    }

    inline Future<> to_future(turbo::Status status) {
        return Future<>::make_finished(std::move(status));
    }

    template<typename T>
    Future<T> to_future(T value) {
        return Future<T>::make_finished(std::move(value));
    }

    template<typename T>
    Future<T> to_future(turbo::Result<T> maybe_value) {
        return Future<T>::make_finished(std::move(maybe_value));
    }

    template<typename T>
    Future<T> to_future(Future<T> fut) {
        return fut;
    }

    template<typename T>
    struct EnsureFuture {
        using type = decltype(to_future(std::declval<T>()));
    };

}  // namespace nebula
